package com.love.cloud.common.redis.queue.consumer;

import com.love.cloud.common.redis.queue.constants.QueueConstant;
import com.love.cloud.common.redis.queue.listener.ConsumerListener;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.stream.StreamListener;

/**
 * @author hongdongqu
 */
public class BaseConsumer implements StreamListener<String, MapRecord<String,String,String>> {

    /**
     * stream名称（主题名称）
     */
    private String streamName = "";
    /**
     * 消费组名
     */
    private String groupName = "";
    /**
     * 消费者名称
     */
    private String consumerName = "";
    /**
     * 消费ack确认机制
     */
    private int consumerAck=1;

    private RedisTemplate redisTemplate;

    private ConsumerListener consumerListener;

    public BaseConsumer(String streamName, String groupName, String consumerName, int consumerAck, RedisTemplate redisTemplate, ConsumerListener consumerListener) {
        this.streamName = streamName;
        this.groupName = groupName;
        this.consumerName = consumerName;
        this.consumerAck = consumerAck;
        this.redisTemplate = redisTemplate;
        this.consumerListener = consumerListener;
    }

    @Override
    public void onMessage(MapRecord<String, String, String> message) {

        StreamOperations<String, String, String> streamOperations = redisTemplate.opsForStream();
        switch (consumerAck){
            case QueueConstant.CONSUMER_ACK_WAIT:
                consumerListener.onMessage(message);
                //保证消息ack机制
                //消息应答
                streamOperations.acknowledge( streamName,groupName,message.getId() );
                break;
            case QueueConstant.CONSUMER_ACK_UN_WAIT:
            default:
                //保证消息ack机制
                //消息应答
                streamOperations.acknowledge( streamName,groupName,message.getId() );
                consumerListener.onMessage(message);

        }
    }
}
